﻿using System;
using System.Net;
using __ts.Core.Api;
using __ts.Core.Service;
using __ts.Core.Service.IdleChecher;
using __ts.Core.Session;
using log4net;

/*
 *  author: able
 *  @2012/2/21 17:46:06
 * 
 */
namespace __ts.Core.Transport.Tcp
{
    /// <summary>
    /// 
    /// </summary>
    public class NioSelectorProcessor : SelectorProcessor
    {
        /** A logger for this class */
        private static ILog LOGGER = LogManager.GetLogger(typeof(NioSelectorProcessor));

        /**
         * A timeout used for the select, as we need to get out to deal with idle
         * sessions
         */
        private static long SELECT_TIMEOUT = 1000L;

        private SelectorStrategy strategy;

        //private Map<SocketAddress, ServerSocketChannel> serverSocketChannels = new ConcurrentHashMap<SocketAddress, ServerSocketChannel>();

        ///** Read buffer for all the incoming bytes (default to 64Kb) */
        //private ByteBuffer readBuffer = ByteBuffer.allocate(64 * 1024);

        /** the thread polling and processing the I/O events */
        private SelectorWorker worker = null;

        /** helper for detecting idleing sessions */
        private IdleChecker idleChecker = new IndexedIdleChecker();

        /** A queue containing the servers to bind to this selector */
        //private final Queue<Object[]> serversToAdd = new ConcurrentLinkedQueue<Object[]>();

        ///** server to remove of the selector */
        //private final Queue<ServerSocketChannel> serversToRemove = new ConcurrentLinkedQueue<ServerSocketChannel>();

        ///** new session freshly accepted, placed here for being added to the selector */
        //private final Queue<NioTcpSession> sessionsToConnect = new ConcurrentLinkedQueue<NioTcpSession>();

        ///** session to be removed of the selector */
        //private final Queue<NioTcpSession> sessionsToClose = new ConcurrentLinkedQueue<NioTcpSession>();

        ///** A queue used to store the sessions to be flushed */
        //private final Queue<NioTcpSession> flushingSessions = new ConcurrentLinkedQueue<NioTcpSession>();

        //private Selector selector;

        //// Lock for Selector worker, using default. can look into fairness later.
        //// We need to think about a lock less mechanism here.
        //private Lock workerLock = new ReentrantLock();

        class ServerSocketChannel { }

        public NioSelectorProcessor(String name, SelectorStrategy strategy)
        {
            this.strategy = strategy;
        }

        /**
         * Add a bound server channel for starting accepting new client connections.
         * 
         * @param serverChannel
         */
        private void add(ServerSocketChannel serverChannel, IoServer server)
        {
            //LOGGER.debug("adding a server channel {} for server {}", serverChannel, server);
            //serversToAdd.add(new Object[] { serverChannel, server });
            wakeupWorker();
        }

        /**
         * Wake the I/O worker thread and if none exists, create a new one
         * FIXME : too much locking there ?
         */
        private void wakeupWorker()
        {
            //workerLock.lock();
            //try {
            //    if (worker == null) {
            //        worker = new SelectorWorker();
            //        worker.start();
            //    }
            //} finally {
            //    workerLock.unlock();
            //}

            //if (selector != null) {
            //    selector.wakeup();
            //}
        }

        /**
         * {@inheritDoc}
         */
        public void bindAndAcceptAddress(IoServer server, SocketAddress address)
        {
            //ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

            //// FIXME : should be "genericified"
            //if (server instanceof AbstractTcpServer) {
            //    serverSocketChannel.socket().setReuseAddress(((AbstractTcpServer) server).isReuseAddress());
            //}
            //serverSocketChannel.socket().bind(address);
            //serverSocketChannel.configureBlocking(false);
            //serverSocketChannels.put(address, serverSocketChannel);
            //add(serverSocketChannel, server);
        }

        /**
         * {@inheritDoc}
         */
        public void unbind(SocketAddress address)
        {
            //ServerSocketChannel channel = serverSocketChannels.get(address);
            //channel.socket().close();
            //channel.close();
            //if (serverSocketChannels.remove(address) == null) {
            //    LOGGER.warn("The server channel for address {} was already unbound", address);
            //}
            //LOGGER.debug("Removing a server channel {}", channel);
            //serversToRemove.add(channel);
            wakeupWorker();
        }

        /**
         * {@inheritDoc}
         */
        public void createSession(IoService service, Object clientSocket)
        {
            //LOGGER.debug("create session");
            //final SocketChannel socketChannel = (SocketChannel) clientSocket;
            //final SocketSessionConfig defaultConfig = (SocketSessionConfig) service.getSessionConfig();
            //final NioTcpSession session = new NioTcpSession(service, socketChannel, strategy.getSelectorForNewSession(this));

            //try {
            //    socketChannel.configureBlocking(false);
            //} catch (IOException e) {
            //    LOGGER.error("Unexpected exception, while configuring socket as non blocking", e);
            //    throw new RuntimeIoException("cannot configure socket as non-blocking", e);
            //}
            //// apply idle configuration
            //session.getConfig().setIdleTimeInMillis(IdleStatus.READ_IDLE,
            //        defaultConfig.getIdleTimeInMillis(IdleStatus.READ_IDLE));
            //session.getConfig().setIdleTimeInMillis(IdleStatus.WRITE_IDLE,
            //        defaultConfig.getIdleTimeInMillis(IdleStatus.WRITE_IDLE));

            //// apply the default service socket configuration
            //Boolean keepAlive = defaultConfig.isKeepAlive();

            //if (keepAlive != null) {
            //    session.getConfig().setKeepAlive(keepAlive);
            //}

            //Boolean oobInline = defaultConfig.isOobInline();

            //if (oobInline != null) {
            //    session.getConfig().setOobInline(oobInline);
            //}

            //Boolean reuseAddress = defaultConfig.isReuseAddress();

            //if (reuseAddress != null) {
            //    session.getConfig().setReuseAddress(reuseAddress);
            //}

            //Boolean tcpNoDelay = defaultConfig.isTcpNoDelay();

            //if (tcpNoDelay != null) {
            //    session.getConfig().setTcpNoDelay(tcpNoDelay);
            //}

            //Integer receiveBufferSize = defaultConfig.getReceiveBufferSize();

            //if (receiveBufferSize != null) {
            //    session.getConfig().setReceiveBufferSize(receiveBufferSize);
            //}

            //Integer sendBufferSize = defaultConfig.getSendBufferSize();

            //if (sendBufferSize != null) {
            //    session.getConfig().setSendBufferSize(sendBufferSize);
            //}

            //Integer trafficClass = defaultConfig.getTrafficClass();

            //if (trafficClass != null) {
            //    session.getConfig().setTrafficClass(trafficClass);
            //}

            //Integer soLinger = defaultConfig.getSoLinger();

            //if (soLinger != null) {
            //    session.getConfig().setSoLinger(soLinger);
            //}

            //// Set the secured flag if the service is to be used over SSL/TLS
            //if (service.isSecured()) {
            //    session.initSecure(service.getSslContext());
            //}

            //// event session created
            //session.processSessionCreated();

            //// add the session to the queue for being added to the selector
            //sessionsToConnect.add(session);
            wakeupWorker();
        }

        /**
         * {@inheritDoc}
         */
        public void flush(AbstractIoSession session)
        {
            //LOGGER.debug("scheduling session {} for writing", session);
            //// add the session to the list of session to be registered for writing
            //flushingSessions.add((NioTcpSession) session);
            // wake the selector for unlocking the I/O thread
            wakeupWorker();
        }

        /**
         * The worker processing incoming session creation, session destruction requests, session write and reads.
         * It will also bind new servers.
         */
        private class SelectorWorker //: Thread
        {
            //// map for finding the keys associated with a given server
            //private Map<ServerSocketChannel, SelectionKey> serverKey = new HashMap<ServerSocketChannel, SelectionKey>();

            //// map for finding read keys associated with a given session
            //private Map<NioTcpSession, SelectionKey> sessionReadKey = new HashMap<NioTcpSession, SelectionKey>();

            public void run()
            {
                //try {
                //    //if (selector == null) {
                //    //    LOGGER.debug("opening a new selector");

                //    //    try {
                //    //        selector = Selector.open();
                //    //    } catch (IOException e) {
                //    //        LOGGER.error("IOException while opening a new Selector", e);
                //    //    }
                //    //}

                //    for (;;) {
                //        try {
                //            // pop server sockets for removing
                //            if (serversToRemove.size() > 0) {
                //                processServerRemove();
                //            }

                //            // pop new server sockets for accepting
                //            if (serversToAdd.size() > 0) {
                //                processServerAdd();
                //            }

                //            // pop new session for starting read/write
                //            if (sessionsToConnect.size() > 0) {
                //                processConnectSessions();
                //            }

                //            // pop session for close, if any
                //            if (sessionsToClose.size() > 0) {
                //                processCloseSessions();
                //            }

                //            LOGGER.debug("selecting...");
                //            int readyCount = selector.select(SELECT_TIMEOUT);
                //            LOGGER.debug("... done selecting : {}", readyCount);

                //            if (readyCount > 0) {
                //                // process selected keys
                //                Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();

                //                // Loop on each SelectionKey and process any valid action
                //                while (selectedKeys.hasNext()) {
                //                    SelectionKey key = selectedKeys.next();
                //                    selectedKeys.remove();

                //                    if (!key.isValid()) {
                //                        continue;
                //                    }

                //                    selector.selectedKeys().remove(key);

                //                    if (key.isAcceptable()) {
                //                        processAccept(key);
                //                    }

                //                    if (key.isReadable()) {
                //                        processRead(key);
                //                    }

                //                    if (key.isWritable()) {
                //                        processWrite(key);
                //                    }

                //                }
                //            }

                //            // registering session with data in the write queue for
                //            // writing
                //            while (!flushingSessions.isEmpty()) {
                //                processFlushSessions();
                //            }
                //        } catch (IOException e) {
                //            LOGGER.error("IOException while selecting selector", e);
                //        }

                //        // stop the worker if needed
                //        workerLock.lock();

                //        try {
                //            if (selector.keys().isEmpty()) {
                //                worker = null;
                //                break;
                //            }
                //        } finally {
                //            workerLock.unlock();
                //        }

                //        // check for idle events
                //        idleChecker.processIdleSession(System.currentTimeMillis());
                //    }
                //} catch (Exception e) {
                //    LOGGER.error("Unexpected exception : ", e);
                //}
            }

            /**
             * Handles the servers removal
             */
            private void processServerRemove()
            {
                //while (!serversToRemove.isEmpty()) {
                //    ServerSocketChannel channel = serversToRemove.poll();
                //    SelectionKey key = serverKey.remove(channel);

                //    if (key == null) {
                //        LOGGER.error("The server socket was already removed of the selector");
                //    } else {
                //        LOGGER.debug("Removing the server from this selector : {}", key);
                //        key.cancel();
                //    }
                //}
            }

            /**
             * Handles the servers addition
             */
            private void processServerAdd()
            {
                //while (!serversToAdd.isEmpty()) {
                //    Object[] tmp = serversToAdd.poll();
                //    ServerSocketChannel channel = (ServerSocketChannel) tmp[0];
                //    SelectionKey key = channel.register(selector, SelectionKey.OP_ACCEPT);
                //    key.attach(tmp);
                //    LOGGER.debug("Accepted the server on this selector : {}", key);
                //}
            }

            /**
             * Handles all the sessions that must be connected
             */
            private void processConnectSessions()
            {
                //while (!sessionsToConnect.isEmpty()) {
                //    NioTcpSession session = sessionsToConnect.poll();
                //    SelectionKey key = session.getSocketChannel().register(selector, SelectionKey.OP_READ);
                //    key.attach(session);
                //    sessionReadKey.put(session, key);

                //    // Switch to CONNECTED, only if the session is not secured, as the SSL Handshake
                //    // will occur later.
                //    if (!session.isSecured()) {
                //        session.setConnected();

                //        // fire the event
                //        ((AbstractIoService) session.getService()).fireSessionCreated(session);
                //        session.processSessionOpened();
                //        long time = System.currentTimeMillis();
                //        idleChecker.sessionRead(session, time);
                //        idleChecker.sessionWritten(session, time);
                //    }
                //}
            }

            /**
             * Handles all the sessions that must be closed
             */
            private void processCloseSessions()
            {
                //while (!sessionsToClose.isEmpty()) {
                //    NioTcpSession session = sessionsToClose.poll();

                //    SelectionKey key = sessionReadKey.remove(session);
                //    key.cancel();

                //    // closing underlying socket
                //    session.getSocketChannel().close();
                //    // fire the event
                //    session.processSessionClosed();
                //    ((AbstractIoService) session.getService()).fireSessionDestroyed(session);
                //}
            }

            /**
             * Processes the Accept action for the given SelectionKey
             */
            private void processAccept(SelectionKey key)
            {
                //LOGGER.debug("acceptable new client {}", key);
                //ServerSocketChannel serverSocket = (ServerSocketChannel) ((Object[]) key.attachment())[0];
                //IoServer server = (IoServer) (((Object[]) key.attachment())[1]);
                //// accepted connection
                //SocketChannel newClientChannel = serverSocket.accept();
                //LOGGER.debug("client accepted");
                //// and give it's to the strategy
                //strategy.getSelectorForNewSession(NioSelectorProcessor.this).createSession(server, newClientChannel);
            }

            /**
             * Processes the Read action for the given SelectionKey
             */
            private void processRead(SelectionKey key)
            {
                //LOGGER.debug("readable client {}", key);
                //NioTcpSession session = (NioTcpSession) key.attachment();
                //SocketChannel channel = session.getSocketChannel();
                //readBuffer.clear();
                //int readCount = channel.read(readBuffer);

                //LOGGER.debug("read {} bytes", readCount);

                //if (readCount < 0) {
                //    // session closed by the remote peer
                //    LOGGER.debug("session closed by the remote peer");
                //    sessionsToClose.add(session);
                //} else {
                //    // we have read some data
                //    // limit at the current position & rewind buffer back to start & push to the chain
                //    readBuffer.flip();

                //    if (session.isSecured()) {
                //        // We are reading data over a SSL/TLS encrypted connection. Redirect
                //        // the processing to the SslHelper class.
                //        SslHelper sslHelper = session.getAttribute(SSL_HELPER, null);

                //        if (sslHelper == null) {
                //            throw new IllegalStateException();
                //        }

                //        sslHelper.processRead(session, readBuffer);
                //    } else {
                //        // Plain message, not encrypted : go directly to the chain
                //        session.processMessageReceived(readBuffer);
                //    }

                //    idleChecker.sessionRead(session, System.currentTimeMillis());
                //}
            }

            /**
             * Processes the Write action for the given SelectionKey
             */
            private void processWrite(SelectionKey key)
            {
                //NioTcpSession session = (NioTcpSession) key.attachment();

                //LOGGER.debug("writable session : {}", session);

                //session.setNotRegisteredForWrite();

                //// write from the session write queue
                //boolean isEmpty = false;

                //try {
                //    Queue<WriteRequest> queue = session.acquireWriteQueue();

                //    do {
                //        // get a write request from the queue
                //        WriteRequest wreq = queue.peek();

                //        if (wreq == null) {
                //            break;
                //        }

                //        ByteBuffer buf = (ByteBuffer) wreq.getMessage();

                //        // Note that if the connection is secured, the buffer already
                //        // contains encrypted data.
                //        int wrote = session.getSocketChannel().write(buf);
                //        session.incrementWrittenBytes(wrote);
                //        LOGGER.debug("wrote {} bytes to {}", wrote, session);

                //        idleChecker.sessionWritten(session, System.currentTimeMillis());

                //        if (buf.remaining() == 0) {
                //            // completed write request, let's remove it
                //            queue.remove();
                //            // complete the future
                //            DefaultWriteFuture future = (DefaultWriteFuture) wreq.getFuture();

                //            if (future != null) {
                //                future.complete();
                //            }
                //        } else {
                //            // output socket buffer is full, we need
                //            // to give up until next selection for
                //            // writing
                //            break;
                //        }
                //    } while (!queue.isEmpty());

                //    isEmpty = queue.isEmpty();
                //} finally {
                //    session.releaseWriteQueue();
                //}

                //// if the session is no more interested in writing, we need
                //// to stop listening for OP_WRITE events
                //if (isEmpty) {
                //    if (session.isClosing()) {
                //        LOGGER.debug("closing session {} have empty write queue, so we close it", session);
                //        // we was flushing writes, now we to the close
                //        session.getSocketChannel().close();
                //    } else {
                //        // a key registered for read ? (because we can have a
                //        // Selector for reads and another for the writes
                //        SelectionKey readKey = sessionReadKey.get(session);

                //        if (readKey != null) {
                //            LOGGER.debug("registering key for only reading");
                //            SelectionKey mykey = session.getSocketChannel().register(selector, SelectionKey.OP_READ,
                //                    session);
                //            sessionReadKey.put(session, mykey);
                //        } else {
                //            LOGGER.debug("cancel key for writing");
                //            session.getSocketChannel().keyFor(selector).cancel();
                //        }
                //    }
                //}
            }

            /**
             * Flushes the sessions
             */
            private void processFlushSessions()
            {
                //NioTcpSession session = flushingSessions.poll();
                //// a key registered for read ? (because we can have a
                //// Selector for reads and another for the writes
                //SelectionKey readKey = sessionReadKey.get(session);

                //if (readKey != null) {
                //    // register for read/write
                //    SelectionKey key = session.getSocketChannel().register(selector,
                //            SelectionKey.OP_READ | SelectionKey.OP_WRITE, session);

                //    sessionReadKey.put(session, key);
                //} else {
                //    session.getSocketChannel().register(selector, SelectionKey.OP_WRITE, session);
                //}
            }
        }

        class SelectionKey { }
    }
}